//
// Created by martin on 3/22/22.
//

#include "FtpServer.h"
#include "FtpContext.h"
#include "FtpResponse.h"
#include "Configurer.h"

#include "muduo/base/Logging.h"
#include "muduo/net/EventLoop.h"

using namespace ftp;
using namespace muduo;
using namespace muduo::net;

using std::placeholders::_1;
using std::placeholders::_2;
using std::placeholders::_3;

int FtpServer::numsOpenConnectionMax_ = -1;


namespace ftp
{
namespace detail
{

// 默认连接数最大值, 从配置文件获取该值
int defaultNumsOpenConnectionMax()
{
	return Configurer::instance().getItemInt("NumsUserMax", 10000);
}

} // namespace detail
} // namespace ftp

FtpServer::FtpServer(EventLoop* loop, const InetAddress& listenAddr, const std::string& name,
	TcpServer::Option option)
	: server_(loop, listenAddr, name, option),
	numsOpenConnection_(0)
{
	server_.setConnectionCallback(std::bind(&FtpServer::onConnection, this, _1));
	server_.setMessageCallback(std::bind(&FtpServer::onMessage, this, _1, _2, _3));
	server_.setWriteCompleteCallback(std::bind(&FtpServer::onWriteComplete, this, _1));

	loop->runEvery(10, std::bind(&FtpServer::onTimer, this)); // 注册定时回调
	// 连接1000s超时, 每个桶100s, 共100个桶
	connectionBuckets_.resize(100);
	//dumpConnectionBuckets();

	cachedNumsOpenConnectionMax();
}

void FtpServer::start()
{
	LOG_INFO << "FtpServer starts listening on " << server_.ipPort();
	server_.start();
}

int FtpServer::getNumsOpenConnection() const
{
	return numsOpenConnection_;
}

// 更新最大用户数, 非线程安全, 必须在FtpServer启动多线程前调用
int FtpServer::cachedNumsOpenConnectionMax()
{
	if (numsOpenConnectionMax_ < 0) {
		numsOpenConnectionMax_ = detail::defaultNumsOpenConnectionMax();
	}
	return numsOpenConnectionMax_;
}

void FtpServer::onConnection(const muduo::net::TcpConnectionPtr& conn)
{
	LOG_INFO << "FtpServer - " << conn->peerAddress().toIpPort() << " -> "
		<< (conn->connected() ? "UP" : "DOWN");

	if (conn->connected()) { // connected
#if 0
		IdleConnectionEntryPtr entry(new IdleConnectionEntry(conn));

		{
			MutexLockGuard lockGuard(mutex_);
			connectionBuckets_.back().insert(entry);
			dumpConnectionBuckets();
		}

		WeakIdleConnectionEntryPtr weakEntry(entry);
#else
		WeakIdleConnectionEntryPtr weakEntry;
#endif

		if (!overNumsOpenConnectionMax()) { // 控制连接数正常
			++numsOpenConnection_;

			std::shared_ptr<FtpContext> context(new FtpContext(conn->getLoop(), conn, weakEntry, fptUserDao_));
			context->startFtpSession();
			conn->setContext(context); // 保存到Control TcpConnection的成员context_
		}
		else { // 控制连接数超阈值
			conn->send("120 Too many users are connected, please try again minutes later...\r\n");
			conn->shutdown();
		}
	}
	else { // disconnected
		--numsOpenConnection_;
#if 0
		FtpContext* context = boost::any_cast<FtpContext>(conn->getMutableContext());
		WeakIdleConnectionEntryPtr weakEntry(context->weakIdleConnectionEntryPtr());
		LOG_DEBUG << "Entry use_count = " << weakEntry.use_count();
#endif
	}
}

void FtpServer::onMessage(const muduo::net::TcpConnectionPtr& conn, muduo::net::Buffer* buffer, muduo::Timestamp receiveTime)
{
	LOG_DEBUG << "FtpServer::onMessage";
	std::shared_ptr<FtpContext> context = boost::any_cast<std::shared_ptr<FtpContext>>(conn->getContext());
	if (!context) {
		LOG_ERROR << "FtpConext object int control connection@" << &conn << " has been null";
		return;
	}

#if 0
	{ // 更新环形缓冲区, 将Entry弱引用提升为强引用, 并插入到缓冲区末尾
		WeakIdleConnectionEntryPtr weakEntry(context->weakIdleConnectionEntryPtr());
		IdleConnectionEntryPtr entry(weakEntry.lock()); // 将弱引用提升为强引用, 插入环形缓冲区
		if (entry) {
			MutexLockGuard lockGuard(mutex_);
			connectionBuckets_.back().insert(entry);
		}
	}
#endif

	// 解析客户端请求
	if (!context->parseRequest(buffer, receiveTime)) {
		conn->send("500 Syntax error, command unrecognized.\r\n");

		if (context->numErrors() > FtpContext::kNumErrorsMax) { // 错误数超阈值
			conn->shutdown();
		}
	}
	// 解析完成
	if (context->gotAll()) {
		context->clearError();
		onRequest(conn, context->request()); // 处理ftp请求
		context->resetRequest();
	}
}

bool FtpServer::addUser(const std::string& username, const std::string& password, const std::string& localRootPath, Permission permission)
{
	return fptUserDao_.addUser(username, password, localRootPath, permission);
}

bool FtpServer::addAnonymousUser(const std::string& localRootPath, Permission permission)
{
	return fptUserDao_.addUser(UserInfo::defaultAnonymousUsername(), "", localRootPath, permission);
}

void FtpServer::onWriteComplete(const muduo::net::TcpConnectionPtr& conn)
{
	LOG_TRACE << "FtpServer::onWriteComplete";
}

/**
 * 由FtpServer::onMessage()回调, 处理ftp command请求. 在解析完一条完整ftp请求后调用
 */
void FtpServer::onRequest(const TcpConnectionPtr& conn, const FtpRequest& req)
{
	LOG_INFO << "FtpServer - " << conn->peerAddress().toIpPort() << " <-- " << req.commandName() << " " << req.commandParam();
	std::shared_ptr<FtpContext> context = boost::any_cast<std::shared_ptr<FtpContext>>(conn->getContext());
	if (!context) {
		LOG_ERROR << "FtpConext object int control connection@" << &conn << " has been null";
		return;
	}

	// handle ftp command request
	context->handleFtpCommand();
}

void FtpServer::onTimer()
{
	connectionBuckets_.push_back(Bucket()); // 周期性往timing wheel末尾插入数据, 队列满时头部元素会移除
//    dumpConnectionBuckets();
}

void FtpServer::dumpConnectionBuckets() const
{
	LOG_DEBUG << "size = " << connectionBuckets_.size();
	int idx = 0;

	for (auto bucketI = connectionBuckets_.begin();
		bucketI != connectionBuckets_.end();
		++bucketI, ++idx) {
		const Bucket& bucket = *bucketI;
		printf("[%d] len = %zd : ", idx, bucket.size());
		for (const auto& it : bucket) {
			bool connectionDead = it->weakConn_.expired();
			printf("%p(%ld)%s, ", get_pointer(it),
				it.use_count(),
				connectionDead ? "DEAD" : "");
		}
		puts("");
	}
}

bool FtpServer::overNumsOpenConnectionMax() const
{
	return cachedNumsOpenConnectionMax() < numsOpenConnection_;
}